package transform

import Source.SensorReading
import org.apache.flink.streaming.api.scala._

object ConnectAndCoMapAndUnion {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val inputPath = "src/main/resources/SensorReading"
    val inputStream = env.readTextFile(inputPath)

    //转换成样例类类型
    val dataStream = inputStream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )
    //分流操作
    val spiltStream = dataStream
      .split(
        data => {
          if (data.temperature > 30.0) Seq("High") else Seq("low")
        }
      )
    //获取流
    val HighTempStream = spiltStream.select("High")
    val LowTempStream = spiltStream.select("low")
    val allStream = spiltStream.select("High", "low")

    //将数据类型转换一下
    val warningStream = HighTempStream.map(data => (data.id, data.temperature))

    val connectStream = warningStream.connect(LowTempStream)

    //CoMap对数据继续处理
    val CoMapStream = connectStream
      .map(
        warningData => (warningData._1, warningData._2, "warning"),
        lowData => (lowData.id, "healthy"
        )
      )
    env.execute()
  }
}
